import os
from contextlib import contextmanager, nullcontext
from mmap import mmap, ACCESS_READ, ACCESS_WRITE
from os import PathLike
from typing import AnyStr, Any, Generator

from rich.progress import (
    Progress,
    TextColumn,
    BarColumn,
    DownloadColumn,
    TransferSpeedColumn,
    TimeElapsedColumn,
    TimeRemainingColumn
)


@contextmanager
def mmap_view(path: AnyStr | PathLike, readonly: bool = True, truncate: int = 0, offset: int = 0, length: int = 0,
              mode: int = 0o644) -> Generator[memoryview, Any, None]:
    flags = os.O_CREAT | os.O_RDWR | os.O_TRUNC if not readonly else os.O_RDONLY
    fd = os.open(path, flags, mode)
    try:
        if not readonly and truncate:
            os.ftruncate(fd, truncate)

        with mmap(fd, length, access=ACCESS_READ if readonly else ACCESS_WRITE, offset=offset) as mm:
            with memoryview(mm) as mv:
                yield mv
    finally:
        os.close(fd)


def common_progress(show_progress=True) -> Progress | nullcontext:
    if not show_progress:
        return nullcontext()

    columns = [
        TextColumn('[progress.description]{task.description}'),
        BarColumn(),
        DownloadColumn(),
        TransferSpeedColumn(),
        TimeElapsedColumn(),
        TimeRemainingColumn(elapsed_when_finished=True),
    ]
    progress = Progress(*columns, auto_refresh=True, refresh_per_second=1)
    return progress


def align_to(i: int, alignment: int) -> int:
    """
    对齐到指定边界
    """
    if alignment <= 0:
        raise ValueError("alignment must be positive")

    return (i + alignment - 1) & ~(alignment - 1)
